package com.chb.storm.gpstrident;

import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.trident.Stream;
import storm.trident.TridentTopology;
import storm.trident.state.StateFactory;
import storm.trident.testing.FixedBatchSpout;
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.AuthorizationException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import com.common.utils.PropUtils;

public class KafkaGpsTrident {

	public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
		String brokerZkStr = PropUtils.getProp("brokerZkStr");
		String brokerZkPath = PropUtils.getProp("brokerZkPath");
		
		ZkHosts zkHosts = new ZkHosts(brokerZkStr, brokerZkPath);
		String topicName = "mytopic";
		
		//redeploy必须保证下面两个字段不变
		String zkRoot = "/kafka_zk";
		String spoutId = "KafkaTopologyHbase20021999";
		
//		SpoutConfig kafkaConfig = new SpoutConfig(zkHosts, topicName, zkRoot, spoutId);
//		kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
//		
//		//忽略在zookeeper中的offset
//		kafkaConfig.ignoreZkOffsets = true;
//		//从当前部署时间来读取topic
//		kafkaConfig.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
		
		Fields fields = new Fields("word", "count");
		
		FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
                new Values("storm", 1),
                new Values("trident", 1),
                new Values("needs", 1),
                new Values("javadoc", 1)
        );
        spout.setCycle(true);

		StateFactory stateFactory = new GpsHbaseStateFactory();
		
		TridentTopology topology = new TridentTopology();
		Stream stream = topology.newStream("spoutToTrident", spout);
//		Stream stream = topology.newStream("spoutToTrident", new KafkaSpout(kafkaConfig));
		
		stream.partitionPersist(stateFactory, fields, new GpsHbaseUpdater());
		
		StormTopology stormTopology = topology.build();
		
		Config conf = new Config();
        conf.setMaxSpoutPending(5);
        conf.setNumWorkers(3);
	    
        StormSubmitter.submitTopology("kafkaGpsTrident", conf, stormTopology);
	}
}
